package com.ibm.cps.graphite;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Throwables;
import com.ibm.interfaces.IMessageClient;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.LocalConfig;
import com.ibm.util.exception.CPSException;
import com.seyren.core.util.config.SeyrenConfig;
import com.seyren.core.util.graphite.GraphiteHttpClient;

import clojure.string__init;

public class GraphiteClient implements IMessageClient {

	private GraphiteHttpClient graphiteHttpClient;
	private static Logger logger;
	private List<String> targets;
	private String from;
	private String end;			
	private Integer time;
	private int limit = 5000;

	public GraphiteClient(String host, int port) throws CPSException {
		try {
			logger = Logger.getLogger(this.getClass());
			PropertyConfigurator.configure(LocalConfig.getLoggerProperties());
			System.setProperty("GRAPHITE_URL", host + ":" + port);

			// ms
			System.setProperty("GRAPHITE_SOCKET_TIMEOUT SETTING", LocalConfig.GRAPHITE_SOCKET_TIMEOUT);
			System.setProperty("GRAPHITE_CONNECTION_REQUEST_TIMEOUT SETTING", LocalConfig.GRAPHITE_CONNECTION_REQUEST_TIMEOUT);
			System.setProperty("GRAPHITE_CONNECT_TIMEOUT SETTING", LocalConfig.GRAPHITE_CONNECT_TIMEOUT);
			SeyrenConfig config = new SeyrenConfig();
			logger.info("Trying to connect to graphite server " + config.getGraphiteHost());
			graphiteHttpClient = new GraphiteHttpClient(config);
			logger.info("GRAPHITE_SOCKET_TIMEOUT SETTING: " + config.getGraphiteSocketTimeout());
			logger.info("GRAPHITE_CONNECTION_REQUEST_TIMEOUT SETTING: " + config.getGraphiteConnectionRequestTimeout());
			logger.info("GRAPHITE_CONNECT_TIMEOUT SETTING: " + config.getGraphiteConnectTimeout());
			targets = Collections.synchronizedList(new ArrayList<String>());
			logger.info("Connected to graphite server " + host + ":" + port + " successfully.");
			
			String interval = LocalConfig.GRAPHITE_INTERVAL;
			String unit = interval.substring(interval.length()-1, interval.length());
			time = Integer.parseInt(interval.substring(0, interval.length() -1));
			from = "-" + 2 * time + unit;
			end = "-" + time + unit;
			if (unit.equals("m")) {
				time = time * 60;
			}
			
			logger.info("interval setting:{" + time +" s}");
			
		} catch (Throwable e) {
			logger.error(Throwables.getStackTraceAsString(e));
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"Can't get connection to graphite server " + host + ":" + port + " successfully.");
		}
	}

	@Override
	public void publishMessage(String destination, String message) throws CPSException {
		publishMessage(destination, message.getBytes());
	}

	@Override
	public void close() {

	}

	@Override
	public void publishMessage(String topicName, byte[] message) throws CPSException {

	}

	@Override
	public synchronized void subscribe(String topicName) throws CPSException {
		if (topicName == null) {
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, "Topicname should not be null.");
		}
		targets.add(topicName);
		
		// to avoid the topicName is too long for exceeding the url length.
		if (topicName.length() <= limit) {
			try {
				graphiteHttpClient.getTargetJson(from, end, topicName);
			} catch (Exception e) {
				logger.error(Throwables.getStackTraceAsString(e));
				throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, e.getMessage());
			}
			logger.info("Subscribe graphite target " + topicName + " successfully.");
		}else{

			// topicName's length > limit , we seperate the topicName into small parts.
			int startIndex = 0;
			int endIndex = limit;
			int length = topicName.length();
			
			topicName += ",";  // to better seperate the str.
			
			while(startIndex < length){
				endIndex = topicName.lastIndexOf(",", endIndex);
				String substr = topicName.substring(startIndex,endIndex);
				String[] targetArray = substr.split(",\\s*");
				logger.debug("try to subscribe small topic " + substr);
				try {
					graphiteHttpClient.getTargetJson(from, end, targetArray);
				} catch (Exception e) {
					logger.error(Throwables.getStackTraceAsString(e));
//					throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, e.getMessage());
				}
				startIndex = endIndex + 1;
				endIndex = startIndex + limit;
			}
			logger.info("Subscribe graphite target " + topicName + " successfully.");
		}


	}

	@Override
	public synchronized void unsubscribe(String topicName) throws CPSException {
		if (!targets.contains(topicName)) {
			logger.warn("Target " + topicName + " hasn't been subscribe before, you should not unsubscribe it.");
			return;
		}
		targets.remove(topicName);
	}

	@Override
	public synchronized Collection<String> consumeMessage() throws CPSException {
		try {
			Collection<String> messages = new ArrayList<String>();
			for (String targetList : targets) {
				if (targetList.length() < limit) {
					logger.debug("======Target: " + targetList);
					String[] targetArray = targetList.split(",\\s*");
					//				JsonNode node = graphiteHttpClient.getTargetJson(from, end, targetArray);
					JsonNode node = graphiteHttpClient.getTargetJson(end, "", targetArray);
					GraphiteDataMessage message = new GraphiteDataMessage(node, targetArray);
					node = message.getJointMessage();
					if (node != null) {
						logger.info("Read message:" + node.toString());
						messages.add(node.toString());
					} 
				}else {

					// topicName's length > limit , we seperate the topicName into small parts.
					int startIndex = 0;
					int endIndex = limit;
					int length = targetList.length();
					
					String[] targetArray = targetList.split(",\\s*");
					
					ArrayNode nodes = JSONUtility.newArrayNode();
					
					targetList += ",";  // to better seperate the str.
					while(startIndex < length){
						endIndex = targetList.lastIndexOf(",", endIndex);
						String substr = targetList.substring(startIndex,endIndex);
						String[] subtargetArray = substr.split(",\\s*");
						logger.debug("======Target: " + targetList);
						//				JsonNode node = graphiteHttpClient.getTargetJson(from, end, targetArray);
						JsonNode subnode = graphiteHttpClient.getTargetJson(end, "", subtargetArray);
						if (subnode.isArray()) {
							for (int i = 0; i < subnode.size(); i++) {
								nodes.add(subnode.get(i));
							}
						}
						startIndex = endIndex + 1;
						endIndex = startIndex + limit;
					}
					
					GraphiteDataMessage message = new GraphiteDataMessage(nodes, targetArray);
					ObjectNode node = message.getJointMessage();
					if (node != null) {
						logger.info("Read message:" + node.toString());
						messages.add(node.toString());
					} 
				
				}
			}
			// modify by gxj
//			TimeUnit.SECONDS.sleep(10);
			TimeUnit.SECONDS.sleep(time);
//			System.out.println(messages);
			return messages;
		} catch (Exception e) {
			logger.error(Throwables.getStackTraceAsString(e));
			// throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
			// e.getMessage());
		}
		return null;
	}
	

}
